package com.thp.bigdata.rjoin.findFriends;

import java.io.IOException;
import java.util.Arrays;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.junit.Test;

public class CommonFriendsStepTwoMapper extends Mapper<LongWritable, Text, Text, Text> {
	@Override
	protected void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		// 传进来的数据 : A   I,K,C,B,G,F,H,O,D,
		String[] friendAndUsers = value.toString().split("\t");
		// friendAndUsers 0 位置为好友  "A"
		String friend = friendAndUsers[0];
		// friendAndUsers 1 位置为拥有上面好友的用户们
		// I K C B G F H O D
		String[] users = friendAndUsers[1].split(",");
		// 将user进行排序，避免重复
		// [B, C, D, F, G, H, I, K, O]
		/**
		 * 【注意：】
		 *   这里是要注意，一定要进行排序，因为我们是使用两次嵌套循环来进行排练组合的
		 *   如果所有的有这个好友的用户的顺序都是乱的，比如说
		 *   A      B,D,C    --  有A好友的是  B C D 三个   [B-D : A]  [B-C : A]  [D-C : A]
		 *   B		C,D,E    --  有B好友的是  C D 两个        [C-D : B]  [D-E : B]
		 *   
		 *   其实看上面的好友对  -- 向reduce输出
		 *    [D-C : A]     [C-D : B]
		 *    上面这两个都是  C 与   D 的共同好友有  A 跟 B
		 *    本来应该写在一起想reduce输出的，归为同一个key
		 *    但是由于没有排序就是两个不一样的key
		 *    那么reduce再输出的时候，就会都输出，所以会造成重复，虽然数据没有错
		 *   
		 */
		Arrays.sort(users);
		// 以  用户 - 好友 为key，好友 作为value传递给reducer
		for(int i = 0; i < users.length -2; i++) {
			for(int j = i+1; j < users.length -1; j++) {
				/**
		     	 * <B-C,A> <B-D,A> <B-F,A> <B-G,A> <B-H,A> <B-I,A> <B-K,A> <B-O,A>
		     	 * <C-D,A> <C-F,A> <C-G,A> <C-H,A> <C-I,A> <C-K,A> <C-O,A>
		     	 * <D-F,A> <D-G,A> <D-H,A> <D-I,A> <D-K,A> <D-O,A>
		     	 * <F-G,A> <F-H,A> <F-I,A> <F-K,A> <F-O,A>
		     	 * <G-H,A> <G-I,A> <G-K,A> <G-O,A>
		     	 * <H-I,A> <H-K,A> <H-O,A>
		     	 * <I-K,A> <I-O,A>
		     	 * <K-O,A>
				 */
				context.write(new Text(users[i] + "-" + users[j]), new Text(friend));
			}
		}
	}
	
	
	
	@Test
	public void test01() {
		String[] arrs = {"I", "K", "C", "B", "G", "F", "H", "O", "D"};
		System.out.println(Arrays.toString(arrs));
		System.out.println("排序：");
		Arrays.sort(arrs);
		System.out.println(Arrays.toString(arrs));
	}
	
	
	
}
